package network.mq;

import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;

import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author 龙小虬
 * @date 2021/5/6 22:51
 */
public class MQHandler extends SimpleChannelInboundHandler {

    // 生产者的缓存报文
    private static LinkedBlockingQueue<String> blockingQueue = new LinkedBlockingQueue<String>();

    // 等待缓存报文的消费者
    private static LinkedBlockingQueue<ChannelHandlerContext> ctxConsumersQueue = new LinkedBlockingQueue<ChannelHandlerContext>();

    // 消费者
    private final static String TYPE_CONSUMER = "consumer";

    // 生产者
    private final static String TYPE_PRODUCER = "producer";

    /**
     * 存放生产者给予的消息，并将消息推送给消费者
     * @param ctx
     * @param o
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object o) throws Exception {
        // 获取到报文信息
        ByteBuf byteBuf = (ByteBuf) o;
        String msg = byteBuf.toString(CharsetUtil.UTF_8);
        JSONObject jsonObject = JSONObject.parseObject(msg);
        String type = jsonObject.getString("type");
        // 生产者
        if(TYPE_PRODUCER.equals(type)){
            JSONObject data = jsonObject.getJSONObject("data");
            blockingQueue.offer(data.toJSONString());
            if(ctxConsumersQueue.size()>0){
                // 说明在生产者在没有推送缓存消息之前，有消费者来获取了缓存消息
                pushMessage();
            }
        }else if(TYPE_CONSUMER.equals(type)){
            // 缓存消息队列中有消息
            if (blockingQueue.size()>0){
                String poll = blockingQueue.poll();
                ctx.writeAndFlush(Unpooled.copiedBuffer(poll, CharsetUtil.UTF_8));
                return;
            }
            // 缓存消息队列中无消息
            ctxConsumersQueue.add(ctx);
        }else {
            throw new IllegalStateException("type类型有问题");
        }
    }

    private void pushMessage() {
        String poll = blockingQueue.poll();
        ctxConsumersQueue.forEach((k)->{
            k.writeAndFlush(Unpooled.copiedBuffer(poll, CharsetUtil.UTF_8));
        });
    }

}